package com.ztesoft.web.datarecovery;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;

import javax.sql.DataSource;

import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.support.JdbcUtils;

import com.ztesoft.core.cache.EhCacheService;
import com.ztesoft.core.db.ds.DynamicDataSourceManager;
import com.ztesoft.core.spring.context.SpringApplicationContextHolder;
import com.ztesoft.framework.log.ZTEsoftLogManager;
import com.ztesoft.framework.util.Utils;
import com.ztesoft.web.baseconfig.db.arg.DmTableRelArg;
import com.ztesoft.web.baseconfig.db.arg.DmTableRelArg.DmTableRelCriteria;
import com.ztesoft.web.baseconfig.db.arg.SubTableDefineArg;
import com.ztesoft.web.baseconfig.db.arg.SubTableDefineArg.SubTableDefineCriteria;
import com.ztesoft.web.baseconfig.db.dao.DmDataStateDao;
import com.ztesoft.web.baseconfig.db.dao.DmSrcTableDao;
import com.ztesoft.web.baseconfig.db.dao.DmStrategyDao;
import com.ztesoft.web.baseconfig.db.dao.DmTableRelDao;
import com.ztesoft.web.baseconfig.db.dao.DmTargetTableDao;
import com.ztesoft.web.baseconfig.db.dao.SubTableDefineDao;
import com.ztesoft.web.baseconfig.db.dao.TargetColMapDao;
import com.ztesoft.web.baseconfig.db.po.DmDataStatePO;
import com.ztesoft.web.baseconfig.db.po.DmSrcTablePO;
import com.ztesoft.web.baseconfig.db.po.DmStrategyPO;
import com.ztesoft.web.baseconfig.db.po.DmTableRelPO;
import com.ztesoft.web.baseconfig.db.po.DmTargetTablePO;
import com.ztesoft.web.baseconfig.db.po.SubTableDefinePO;
import com.ztesoft.web.baseconfig.db.po.TargetColMapPO;
import com.ztesoft.web.common.DMSConstant;
import com.ztesoft.web.common.customdb.SQLHelper;
import com.ztesoft.web.common.db.dao.DbBaseInfoDao;
import com.ztesoft.web.common.db.dao.DmDao;
import com.ztesoft.web.common.db.po.DbBaseInfoPO;
import com.ztesoft.web.common.db.po.DmPO;
import com.ztesoft.web.common.exchange.ExchangeFunc;
import com.ztesoft.web.datarecovery.db.po.SubTableSqlInfoPO;
import com.ztesoft.web.syncbaseinfo.service.BaseInfoService;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskDao;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskStageDao;
import com.ztesoft.web.taskgenerator.db.po.DmTaskPO;
import com.ztesoft.web.taskgenerator.db.po.DmTaskStagePO;

/**
 * 数据恢复处理线程 Created by linjie on 2016/4/12.
 */
public class DataRecoveryThread implements Callable<String> {
	private static final ZTEsoftLogManager logger = ZTEsoftLogManager.getLogger(DataRecoveryThread.class);

	Integer dmTaskId;// 任务ID

	private DmTaskDao dmTaskDao;
	private DmDataStateDao dmDataStateDao;
	private DmTaskStageDao dmTaskStageDao;
	private DmDao dmDao;
	private DmStrategyDao dmStrategyDao;
	private DmSrcTableDao dmSrcTableDao;
	private DmTargetTableDao dmTargetTableDao;
	private TargetColMapDao targetColMapDao;
	private DmTableRelDao dmTableRelDao;
	private SubTableDefineDao subTableDefineDao;
	private DbBaseInfoDao dbBaseInfoDao;

	private EhCacheService dmsCacheService;// 缓存
	private BaseInfoService baseInfoService;

	private DynamicDataSourceManager dynamicDataSourceManager;// 数据库连接

	private ExchangeFunc exchangeFunc;// 转换方法

	private DmTaskPO dtPO;

	public DataRecoveryThread(Integer dmTaskId_get) {
		dmTaskId = dmTaskId_get;

		dmTaskDao = SpringApplicationContextHolder.getBean(DmTaskDao.class);
		dmDataStateDao = SpringApplicationContextHolder.getBean(DmDataStateDao.class);
		dmTaskStageDao = SpringApplicationContextHolder.getBean(DmTaskStageDao.class);
		dmDao = SpringApplicationContextHolder.getBean(DmDao.class);
		dmStrategyDao = SpringApplicationContextHolder.getBean(DmStrategyDao.class);
		dmSrcTableDao = SpringApplicationContextHolder.getBean(DmSrcTableDao.class);
		dmTargetTableDao = SpringApplicationContextHolder.getBean(DmTargetTableDao.class);
		targetColMapDao = SpringApplicationContextHolder.getBean(TargetColMapDao.class);
		dmTableRelDao = SpringApplicationContextHolder.getBean(DmTableRelDao.class);
		subTableDefineDao = SpringApplicationContextHolder.getBean(SubTableDefineDao.class);
		dbBaseInfoDao = SpringApplicationContextHolder.getBean(DbBaseInfoDao.class);

		dmsCacheService = SpringApplicationContextHolder.getBean(EhCacheService.class);// 缓存
		baseInfoService = SpringApplicationContextHolder.getBean(BaseInfoService.class);

		dynamicDataSourceManager = SpringApplicationContextHolder.getBean(DynamicDataSourceManager.class);// 数据库连接

		exchangeFunc = SpringApplicationContextHolder.getBean(ExchangeFunc.class);
	}

	private void init() {
		// 用任务ID匹配内存中的计划状态数据，判断是否允许进行数据恢复
		dtPO = dmTaskDao.selectByPrimaryKey(dmTaskId);
	}

	@Override
	public String call() {
		try {
			init();
			return doDataRecovery();
		} catch (Exception e) {
			e.printStackTrace();
			logger.error("数据恢复失败，任务ID=" + dmTaskId, e);
			return DMSConstant.Status.FALSE.toString();
		}
	}

	public String doDataRecovery() {
		logger.info("数据恢复开始，任务ID=" + dmTaskId);
		// 时间格式化
		DateFormat format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
		// Integer dmPlanState_dataRecovery = (Integer)
		// dmsCacheService.get("dmsCache",
		// "dataRecovery_" + dtPO.getDmPlanId());
		Integer dmPlanState_dataRecovery = dmDataStateDao.selectByDmTaskId(dmTaskId).getState();
		// if (dmPlanState_dataRecovery == 0 || dmPlanState_dataRecovery == 4) {
		// if (true) {
		// 更新任务状态和DM_DATA_STATE为数据恢复中
		logger.info("更新任务状态和DM_DATA_STATE为数据恢复中，任务ID=" + dmTaskId);
		dtPO.setState("0DS");
		dtPO.setStateDate(new Date());
		dmTaskDao.updateByPrimaryKeySelective(dtPO);
		DmDataStatePO ddsPO = new DmDataStatePO();
		ddsPO.setDmPlanId(dtPO.getDmPlanId());
		ddsPO.setDmTaskId(dmTaskId);
		ddsPO.setState(1);
		ddsPO.setStateDate(new Date());
		dmDataStateDao.updateByPrimaryKeySelective(ddsPO);
		// 更新了dm_data_state记录后，要同步更新其在内存中静态资料
		logger.info("更新了dm_data_state记录后，要同步更新其在内存中静态资料，任务ID=" + dmTaskId);
		dmsCacheService.put("dmsCache", "dataRecovery_" + dtPO.getDmPlanId(), 1);

		// 根据任务id提取任务环节表清理环节记录
		// 只取“数据恢复中”、“处理完成”或“失败”的清理环节记录
		logger.info("根据任务id提取任务环节表清理环节记录，只取“数据恢复中”、“处理完成”或“失败”的清理环节记录，任务ID=" + dmTaskId);
		List<DmTaskStagePO> dtsPOList = dmTaskStageDao.selectForDataRecovery(dmTaskId);
		for (DmTaskStagePO dtsPO : dtsPOList) {
			// 获取源表信息，用于判断是否是从表
			DmSrcTablePO dstPO = dmSrcTableDao.selectByPrimaryKey(dtsPO.getSourceDataId());
			// 从表不处理
			if (Utils.notEmpty(dstPO.getMainDataId())) {// 从表
				continue;
			}

			// 获取该主表下的从表的数据库执行sql
			List<SubTableSqlInfoPO> subTableSqlInfos = getSubTableSQL(dtsPOList, dtsPO.getSourceDataId());

			// 更新其状态为数据恢复中
			dtsPO.setState("0DS");
			dtsPO.setStateDate(new Date());
			dmTaskStageDao.updateByPrimaryKeySelective(dtsPO);

			// 获取目标表的信息：ID、数据库连接ID、数据库用户名
			DmStrategyPO dsPO = dmStrategyDao.selectByPrimaryKey(dtsPO.getDmStrategyId());
			DmTargetTablePO dttPO = dmTargetTableDao.selectBySourceId(dtsPO.getSourceDataId());

			// 主表源表具体表名
			String checkObjName = dtsPO.getCheckObjName();

			// 先判断是否存在dm_XXX_his表
			int count_his = dmTaskDao.selectExistTableName(("DM_" + checkObjName + "_HIS").toLowerCase());
			if (count_his == 0) {// 不存在
				logger.info(Thread.currentThread() + "DM_" + checkObjName + "_HIS" + "表不存在，说明没有数据");
				// 更新环节记录状态为已恢复
				logger.info("更新环节记录状态为已恢复，任务ID=" + dmTaskId);
				DmTaskStagePO dtsPO_update = new DmTaskStagePO();
				dtsPO_update.setDmInfoId(dtsPO.getDmInfoId());
				dtsPO_update.setState("00S");
				dtsPO_update.setStateDate(new Date());
				dmTaskStageDao.updateByPrimaryKeySelective(dtsPO_update);
				continue;
			}

			// 用环节表的源表名找出对应的任务结果明细历史表，把任务结果明细历史表移到任务结果恢复表
			// 判断表名是否存在，如果存在则不做创建表动作，不存在再创建一张(R_DM_表名)的体检结果明细表
			logger.info(Thread.currentThread()
					+ "用环节表的源表名找出对应的任务结果明细历史表：DM_表名_his，把任务结果明细历史表移到任务结果恢复表，判断表名是否存在，如果存在则不做创建表动作，不存在再创建一张(R_DM_表名)的体检结果明细表，表名：R_DM_"
					+ checkObjName + "，任务ID=" + dmTaskId);
			int count = dmTaskDao.selectExistTableName(("R_DM_" + checkObjName).toLowerCase());
			if (count == 0) {// 创建一张R_DM_tableName表
				dmTaskDao.addCreateRDmTable(checkObjName);
			} else {// 直接把数据移动
				dmTaskDao.addRDmTableDate(checkObjName);
			}
			// 获取需要恢复数据的目标表名，逐一恢复数据，确定需要恢复的目标表的个数以及表名
			logger.info("获取需要恢复数据的目标表名，逐一恢复数据，确定需要恢复的目标表的个数以及表名，表名：R_DM_" + checkObjName + "，任务ID=" + dmTaskId);
			List<DmPO> dmPOList = dmDao.selectRDTargetTableName("R_DM_" + checkObjName, dmTaskId);
			logger.info("结果明细表中的目标表的数据量：" + dmPOList.size() + "， 表名：R_DM_" + checkObjName + "，任务ID=" + dmTaskId);
			for (DmPO dmPO : dmPOList) {
				// 1、数据恢复
				// 目标数据的dm_detail_id，用于更改本地的R_DM_xxx表的状态
				List<Integer> dmDetailIds = new ArrayList<Integer>();
				try {
					// 开始恢复数据
					// 连接目标表数据库
					logger.info("连接目标表数据库，任务ID=" + dmTaskId + "，DbLinkId：" + dttPO.getDbLinkId() + "，DbUserName："
							+ dttPO.getDbUserName());
					String[] targetTableConfig;
					targetTableConfig = baseInfoService.getDbConfig(dttPO.getDbLinkId(), dttPO.getDbUserName());
					DataSource targetTableDs = dynamicDataSourceManager.getDruidDataSource(
							dttPO.getDbLinkId() + dttPO.getDbUserName(), targetTableConfig);
					Connection targetTableCon = DataSourceUtils.getConnection(targetTableDs);
					if (DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(targetTableConfig[0].toUpperCase())) {
						targetTableCon.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
					}
					// 获取目标数据库类型
					DbBaseInfoPO dbiPO = dbBaseInfoDao.selectForDbType(dttPO.getDbLinkId());
					String dbType = dbiPO.getDbCategory();
					// 拼接目标表的sql语句
					// String targetTableSql = "select * from " +
					// dmPO.getDmTargetTableName() + " where dm_task_id = "
					// + dmTaskId + " limit " + dsPO.getBatchCommitCount();
					logger.info("获取目标数据库类型，任务ID=" + dmTaskId + "，type：" + dbType);
					String targetTableSql = SQLHelper.selectByCondition("*", dmPO.getDmTargetTableName(),
							dstPO.getSelectCondition() + " and " + "dm_task_id = " + dmTaskId,
							dsPO.getBatchCommitCount() + "", dbType);
					logger.info("拼接目标表的sql语句，任务ID=" + dmTaskId + "，sql：" + targetTableSql);
					ResultSet targetTableRs = null;
					Boolean isOK = false;
					while (!isOK) {
						// 清空dmDetailIds
						dmDetailIds.clear();
						// 获取目标表数据
						logger.info("获取目标表数据，任务ID=" + dmTaskId);
						PreparedStatement targetTablePstmt = targetTableCon.prepareStatement(targetTableSql.toString(),
								ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
						// PreparedStatement targetTablePstmt =
						// targetTableCon.prepareStatement(targetTableSql
						// .toString());
						targetTableRs = targetTablePstmt.executeQuery();
						ResultSetMetaData targetTableRsmd = targetTableRs.getMetaData();
						int targetTableColumns = targetTableRsmd.getColumnCount();
						// 判断还有没有目标数据
						if (Utils.notEmpty(targetTableRs) && !targetTableRs.next()) {
							isOK = true;
							break;
						}
						// 如果还要用结果集，就把指针再移到初始化的位置
						targetTableRs.beforeFirst();

						// 连接源表数据库
						logger.info("连接源表数据库，任务ID=" + dmTaskId + "，DbLinkId：" + dstPO.getDbLinkId() + "，DbUserName："
								+ dstPO.getDbUserName());
						String[] srcTableConfig = baseInfoService.getDbConfig(dstPO.getDbLinkId(),
								dstPO.getDbUserName());
						DataSource srcTableDs = dynamicDataSourceManager.getDruidDataSource(
								dstPO.getDbLinkId() + dstPO.getDbUserName(), srcTableConfig);
						Connection srcTableCon = DataSourceUtils.getConnection(srcTableDs);
						if (DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(srcTableConfig[0].toUpperCase())) {
							srcTableCon.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
						}

						// 获取源表与目标表的字段映射关系
						TargetColMapPO tcmPO = new TargetColMapPO();
						tcmPO.setSourceDataId(dstPO.getSourceDataId());
						tcmPO.setTargetDataId(dttPO.getTargetDataId());
						List<TargetColMapPO> tcmPOList = targetColMapDao.selectByDoubleId(tcmPO);

						// 拼接插入源表数据sql
						StringBuffer srcTableSql = new StringBuffer();
						srcTableSql.append("insert into " + dmPO.getDmObjectName() + "(");
						StringBuffer srcTableSql_colValue = new StringBuffer();
						// 拼接删除目标表数据sql
						StringBuffer targetTableDeleteSql = new StringBuffer();
						targetTableDeleteSql.append("delete from " + dmPO.getDmTargetTableName() + " where 1=1");

						// 目标表需要恢复的字段的序号
						// {"1,columnName","2,columnName",...}
						List<String> columnsIndex = new ArrayList<String>();
						// 主表目标表与从表的关联字段的序号
						List<Integer> subRelColumnsIndex = new ArrayList<Integer>();

						// 目标表的需要删除的“主键”的序号
						// {"1,columnName","2,columnName",...}
						List<String> columnsIndex_delete = new ArrayList<String>();
						// 目标表“主键”
						List<String> colunms_delete_list = new ArrayList<String>();
						String colunms_delete[] = dstPO.getKeyColumn().split(",");
						for (int i = 0; i < colunms_delete.length; i++) {
							for (TargetColMapPO po : tcmPOList) {
								if (po.getSourceColumn().toLowerCase().equals(colunms_delete[i].toLowerCase())) {
									colunms_delete_list.add(colunms_delete[i].toLowerCase());
								}
							}
						}

						// 目标表字段
						for (int i = 1; i <= targetTableColumns; i++) {
							for (int j = 0; j < colunms_delete_list.size(); j++) {
								if (colunms_delete_list.get(j).toLowerCase()
										.equals(targetTableRsmd.getColumnName(i).toLowerCase())) {// 目标表需要删除的“主键”
									columnsIndex_delete.add(i + "," + colunms_delete_list.get(j).toLowerCase());
									targetTableDeleteSql.append(" and " + colunms_delete_list.get(j).toLowerCase()
											+ " = ?");
								}
							}
							for (TargetColMapPO po : tcmPOList) {
								if (po.getTargetColumn().toLowerCase()
										.equals(targetTableRsmd.getColumnName(i).toLowerCase())) {// 需要恢复
									columnsIndex.add(i + "," + po.getTargetColumn().toLowerCase() + ","
											+ po.getSourceColumnType());
									srcTableSql.append(po.getSourceColumn() + ",");
									srcTableSql_colValue.append("?,");
									break;
								}
							}
							// 拼接从表的select语句
							for (SubTableSqlInfoPO stsiPO : subTableSqlInfos) {
								for (int j = 0; j < stsiPO.getMainTargetColumns().size(); j++) {
									if (stsiPO.getMainTargetColumns().get(j).toLowerCase()
											.equals(targetTableRsmd.getColumnName(i).toLowerCase())) {
										subRelColumnsIndex.add(i);
										stsiPO.setSelect3(stsiPO.getSelect3() + " and b."
												+ targetTableRsmd.getColumnName(i) + " = ?");
									}
								}
							}
						}
						srcTableSql.deleteCharAt(srcTableSql.length() - 1);
						srcTableSql.append(")values(");
						srcTableSql_colValue.deleteCharAt(srcTableSql_colValue.length() - 1);
						srcTableSql.append(srcTableSql_colValue.toString() + ")");
						logger.info("拼接插入源表数据sql，任务ID=" + dmTaskId + "，sql：" + srcTableSql);
						logger.info("拼接删除目标表数据sql，任务ID=" + dmTaskId + "，sql：" + targetTableDeleteSql);

						// 拼接从表的select语句
						for (SubTableSqlInfoPO stsiPO : subTableSqlInfos) {
							stsiPO.setSelect3(dmPO.getDmTargetTableName() + stsiPO.getSelect3());
						}

						// 把要恢复的数据放到sql里
						PreparedStatement srcTablePstmt = srcTableCon.prepareStatement(srcTableSql.toString());
						targetTablePstmt = targetTableCon.prepareStatement(targetTableDeleteSql.toString());
						while (targetTableRs.next()) {
							// 记录dm_detail_id
							dmDetailIds.add(targetTableRs.getInt("dm_detail_id"));
							// 向源表、目标表的sql插入数据
							for (int i = 0; i < columnsIndex.size(); i++) {
								if ("oracle".equals(dbType.toLowerCase())
										&& (Integer.parseInt(columnsIndex.get(i).split(",")[2]) == Types.DATE
												|| Integer.parseInt(columnsIndex.get(i).split(",")[2]) == Types.TIMESTAMP || Integer
												.parseInt(columnsIndex.get(i).split(",")[2]) == Types.TIME)) {
									srcTablePstmt.setTimestamp(i + 1, Timestamp.valueOf(targetTableRs.getString(Integer
											.parseInt(columnsIndex.get(i).split(",")[0]))));
								} else {
									srcTablePstmt.setObject(i + 1, targetTableRs.getString(Integer
											.parseInt(columnsIndex.get(i).split(",")[0])));
								}
							}

							// 向删除的sql插入数据
							for (int i = 0; i < columnsIndex_delete.size(); i++) {
								if ("oracle".equals(dbType.toLowerCase())
										&& (targetTableRs.getType() == Types.DATE
												|| targetTableRs.getType() == Types.TIMESTAMP || targetTableRs
												.getType() == Types.TIME)) {
									targetTablePstmt.setTimestamp(i + 1, Timestamp.valueOf(targetTableRs
											.getString(Integer.parseInt(columnsIndex_delete.get(i).split(",")[0]))));
								} else {
									targetTablePstmt.setObject(
											i + 1,
											targetTableRs.getString(Integer.parseInt(columnsIndex_delete.get(i).split(
													",")[0])));
								}
							}

							// 查询从表数据，并向从表的源表、目标表的sql插入数据
							setSubTableData(targetTableRs, subRelColumnsIndex, subTableSqlInfos, columnsIndex);

							// 源表插入语句加入批次
							srcTablePstmt.addBatch();
							// 目标表删除语句加入批次
							targetTablePstmt.addBatch();
						}
						logger.info("源表插入语句统一提交，任务ID=" + dmTaskId);
						srcTablePstmt.executeBatch();
						JdbcUtils.closeStatement(srcTablePstmt);
						// 释放源表的数据库连接
						DataSourceUtils.releaseConnection(srcTableCon, srcTableDs);

						logger.info("目标表删除语句统一提交，任务ID=" + dmTaskId);
						targetTablePstmt.executeBatch();
						JdbcUtils.closeStatement(targetTablePstmt);
						JdbcUtils.closeResultSet(targetTableRs);

						// 3、更新任务结果恢复明细记录状态
						if (dmDetailIds.size() > 0) {
							logger.info("更新任务结果恢复明细记录状态，任务ID=" + dmTaskId);
							String dateStr = format.format(new Date());
							StringBuffer dmDetailIdStr = new StringBuffer();
							for (int i = 0; i < dmDetailIds.size(); i++) {
								dmDetailIdStr.append(dmDetailIds.get(i) + ",");
							}
							dmDetailIdStr.deleteCharAt(dmDetailIdStr.length() - 1);
							dmTaskDao.updateRD("R_DM_" + checkObjName, "'" + dateStr + "'", dmDetailIdStr.toString());
						}
					}
					// 释放目标表的数据库连接
					DataSourceUtils.releaseConnection(targetTableCon, targetTableDs);

				} catch (Exception e) {
					e.printStackTrace();
					// 数据恢复失败
					logger.error("数据恢复失败，任务ID=" + dmTaskId, e);
					// 更新任务结果恢复明细表
					if (dmDetailIds.size() > 0) {
						logger.info("更新任务结果恢复明细表-失败，任务ID=" + dmTaskId);
						String dateStr = format.format(new Date());
						StringBuffer dmDetailIdStr = new StringBuffer();
						for (int i = 0; i < dmDetailIds.size(); i++) {
							dmDetailIdStr.append(dmDetailIds.get(i) + ",");
						}
						dmDetailIdStr.deleteCharAt(dmDetailIdStr.length() - 1);
						dmTaskDao.updateRDError("R_DM_" + checkObjName, "'" + dateStr + "'", dmDetailIdStr.toString());
					}
					// 更新任务环节表
					logger.info("更新任务环节表-失败，任务ID=" + dmTaskId);
					DmTaskStagePO dtsPO_update = new DmTaskStagePO();
					dtsPO_update.setDmInfoId(dtsPO.getDmInfoId());
					dtsPO_update.setState("0SF");
					dtsPO_update.setStateDate(new Date());
					dmTaskStageDao.updateByPrimaryKeySelective(dtsPO_update);
					// 更新任务表
					logger.info("更新任务表-失败，任务ID=" + dmTaskId);
					DmTaskPO dtPO_update = new DmTaskPO();
					dtPO_update.setDmTaskId(dtPO.getDmTaskId());
					dtPO_update.setState("0SF");
					dtPO_update.setStateDate(new Date());
					dmTaskDao.updateByPrimaryKeySelective(dtPO_update);
					// 更新恢复状态表
					logger.info("更新恢复状态表-失败，任务ID=" + dmTaskId);
					DmDataStatePO ddsPO_update = new DmDataStatePO();
					ddsPO_update.setDmPlanId(ddsPO.getDmPlanId());
					ddsPO_update.setState(4);
					ddsPO_update.setStateDate(new Date());
					dmDataStateDao.updateByPrimaryKeySelective(ddsPO_update);
					// 更新了dm_data_state记录后，要同步更新其在内存中静态资料
					logger.info("更新了dm_data_state记录后，要同步更新其在内存中静态资料-失败，任务ID=" + dmTaskId);
					dmsCacheService.put("dmsCache", "dataRecovery_" + dtPO.getDmPlanId(), 4);
					return "0";
				}
			}
			// 更新环节记录状态为已恢复
			logger.info("更新环节记录状态为已恢复，任务ID=" + dmTaskId);
			DmTaskStagePO dtsPO_update = new DmTaskStagePO();
			dtsPO_update.setDmInfoId(dtsPO.getDmInfoId());
			dtsPO_update.setState("00S");
			dtsPO_update.setStateDate(new Date());
			dmTaskStageDao.updateByPrimaryKeySelective(dtsPO_update);
		}
		// 恢复完成，更新任务表状态为恢复完成00S，流程结束
		logger.info("恢复完成，更新任务表状态为恢复完成00S，流程结束，任务ID=" + dmTaskId);
		dtPO.setState("00S");
		dtPO.setStateDate(new Date());
		dmTaskDao.updateByPrimaryKeySelective(dtPO);
		ddsPO.setState(2);
		ddsPO.setStateDate(new Date());
		dmDataStateDao.updateByPrimaryKeySelective(ddsPO);
		// 更新了dm_data_state记录后，要同步更新其在内存中静态资料
		logger.info("更新了dm_data_state记录后，要同步更新其在内存中静态资料，任务ID=" + dmTaskId);
		dmsCacheService.put("dmsCache", "dataRecovery_" + dtPO.getDmPlanId(), 2);
		// }
		logger.info("恢复完成，流程结束，任务ID=" + dmTaskId);
		return "1";
	}

	/**
	 * 获取该主表下的从表的数据库执行sql
	 * 
	 * @param dtsPOList
	 * @param sourceDataId
	 * @return
	 */
	private List<SubTableSqlInfoPO> getSubTableSQL(List<DmTaskStagePO> dtsPOList, Integer sourceDataId) {
		List<SubTableSqlInfoPO> subTableSQLInfos = new ArrayList<SubTableSqlInfoPO>();
		for (DmTaskStagePO dtsPO : dtsPOList) {
			// 获取源表信息，用于查找从表
			DmSrcTablePO dstPO = dmSrcTableDao.selectByPrimaryKey(dtsPO.getSourceDataId());
			// 查找该主表的从表
			if (Utils.notEmpty(dstPO.getMainDataId()) && sourceDataId.equals(dstPO.getMainDataId())) {
				// 存放从表相关sql的PO
				SubTableSqlInfoPO subTableSQLInfo = new SubTableSqlInfoPO();

				// 获取目标表，策略表信息
				DmStrategyPO dsPO = dmStrategyDao.selectByPrimaryKey(dtsPO.getDmStrategyId());
				DmTargetTablePO dttPO = dmTargetTableDao.selectBySourceId(dtsPO.getSourceDataId());

				// 获取主从表源表的字段映射关系
				DmTableRelArg dtrArg = new DmTableRelArg();
				DmTableRelCriteria dtrCriteria = dtrArg.createCriteria();
				dtrCriteria.andSubDataIdEqualTo(dstPO.getSourceDataId());
				dtrCriteria.andRelDataIdEqualTo(dstPO.getMainDataId());
				DmTableRelPO dtrPO = dmTableRelDao.selectByArg(dtrArg).get(0);
				String mainSrcColumns[] = dtrPO.getRelMapCols().split(",");
				String subSrcColumns[] = dtrPO.getSubMapCols().split(",");

				// 匹配主从表目标表的字段映射关系
				// 主表目标表映射字段
				TargetColMapPO tcmPO_main = new TargetColMapPO();
				tcmPO_main.setSourceDataId(dstPO.getMainDataId());
				// 获取主表目标表ID
				DmTargetTablePO dttPO_main = dmTargetTableDao.selectBySourceId(dstPO.getMainDataId());
				tcmPO_main.setTargetDataId(dttPO_main.getTargetDataId());
				List<TargetColMapPO> tcmPOList_main = targetColMapDao.selectByDoubleId(tcmPO_main);
				// 从表目标表映射字段
				TargetColMapPO tcmPO_sub = new TargetColMapPO();
				tcmPO_sub.setSourceDataId(dstPO.getSourceDataId());
				// 获取从表目标表ID
				DmTargetTablePO dttPO_sub = dmTargetTableDao.selectBySourceId(dstPO.getSourceDataId());
				tcmPO_sub.setTargetDataId(dttPO_sub.getTargetDataId());
				List<TargetColMapPO> tcmPOList_sub = targetColMapDao.selectByDoubleId(tcmPO_sub);

				// 匹配主表目标字段
				List<String> mainTargetColumns = new ArrayList<String>();
				for (TargetColMapPO tcmPO : tcmPOList_main) {
					for (int i = 0; i < mainSrcColumns.length; i++) {
						if (tcmPO.getSourceColumn().equals(mainSrcColumns[i])) {
							mainTargetColumns.add(tcmPO.getTargetColumn());
						}
					}
				}
				// 匹配从表目标字段
				List<String> subTargetColumns = new ArrayList<String>();
				for (TargetColMapPO tcmPO : tcmPOList_sub) {
					for (int i = 0; i < subSrcColumns.length; i++) {
						if (tcmPO.getSourceColumn().equals(subSrcColumns[i])) {
							subTargetColumns.add(tcmPO.getTargetColumn());
						}
					}
				}

				// 拼接where条件
				String subTargetSql_where = "";
				if (mainTargetColumns.size() != 0) {
					for (int i = 0; i < mainTargetColumns.size(); i++) {
						subTargetSql_where += " and a." + mainTargetColumns.get(i) + " = b.";
						subTargetSql_where += subTargetColumns.get(i);
					}
				}

				// 拼接从表目标表数据查询sql
				String subTargetSql1 = "select a.* from ";
				String subTargetSql2 = " as a, ";
				String subTargetSql3 = " as b" + " where a.dm_task_id = " + dmTaskId + subTargetSql_where;
				String subTargetSql4 = " limit " + dsPO.getBatchCommitCount();

				// 获取源表与目标表的字段映射关系
				TargetColMapPO tcmPO = new TargetColMapPO();
				tcmPO.setSourceDataId(dstPO.getSourceDataId());
				tcmPO.setTargetDataId(dttPO.getTargetDataId());
				List<TargetColMapPO> tcmPOList = targetColMapDao.selectByDoubleId(tcmPO);

				subTableSQLInfo.setTcmPOList(tcmPOList);

				subTableSQLInfo.setDmStrategyId(dtsPO.getDmStrategyId());
				subTableSQLInfo.setSourceDataId(dtsPO.getSourceDataId());
				subTableSQLInfo.setDmInfoId(dtsPO.getDmInfoId());

				subTableSQLInfo.setSelect1(subTargetSql1);
				subTableSQLInfo.setSelect2(subTargetSql2);
				subTableSQLInfo.setSelect3(subTargetSql3);
				subTableSQLInfo.setSelect4(subTargetSql4);

				subTableSQLInfo.setMainTargetColumns(mainTargetColumns);
				subTableSQLInfo.setSubTargetColumns(subTargetColumns);

				subTableSQLInfos.add(subTableSQLInfo);
			}
		}
		return subTableSQLInfos;
	}

	/**
	 * 查询从表数据，并向从表的源表、目标表的sql插入数据
	 * 
	 * @param targetTableRs
	 * @param subRelColumnsIndex
	 * @param subTableSqlInfos
	 * @param columnsIndex_main
	 */
	private void setSubTableData(ResultSet targetTableRs, List<Integer> subRelColumnsIndex,
			List<SubTableSqlInfoPO> subTableSqlInfos, List<String> columnsIndex_main) {
		for (SubTableSqlInfoPO stsiPO : subTableSqlInfos) {
			logger.info("查询从表数据，并向从表的源表、目标表的sql插入数据，任务ID=" + dmTaskId);
			// 获取源表、目标表的信息：ID、数据库连接ID、数据库用户名
			DmSrcTablePO dstPO = dmSrcTableDao.selectByPrimaryKey(stsiPO.getSourceDataId());
			DmTargetTablePO dttPO = dmTargetTableDao.selectBySourceId(stsiPO.getSourceDataId());

			try {
				// 从表源表表名
				stsiPO.setSubSrcTableName(getTableName(dstPO.getSourceDataId(), stsiPO, columnsIndex_main,
						targetTableRs));
				// 从表目标表表名
				stsiPO.setSubTargetTableName(getTableName(dttPO.getTargetDataId(), stsiPO, columnsIndex_main,
						targetTableRs));

				// 连接从表目标表数据库
				logger.info("连接从表目标表数据库，任务ID=" + dmTaskId + "，DbLinkId：" + dttPO.getDbLinkId() + "，DbUserName："
						+ dttPO.getDbUserName());
				String[] subTargetTableConfig;
				subTargetTableConfig = baseInfoService.getDbConfig(dttPO.getDbLinkId(), dttPO.getDbUserName());
				DataSource targetTableDs = dynamicDataSourceManager.getDruidDataSource(
						dttPO.getDbLinkId() + dttPO.getDbUserName(), subTargetTableConfig);
				Connection targetTableCon = DataSourceUtils.getConnection(targetTableDs);
				if (DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(subTargetTableConfig[0].toUpperCase())) {
					targetTableCon.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
				}
				PreparedStatement targetPsmt_select = targetTableCon.prepareStatement(stsiPO.getSelect1()
						+ stsiPO.getSubTargetTableName() + stsiPO.getSelect2() + stsiPO.getSelect3()
						+ stsiPO.getSelect4());

				// 连接从表源表数据库
				logger.info("连接从表源表数据库，任务ID=" + dmTaskId + "，DbLinkId：" + dstPO.getDbLinkId() + "，DbUserName："
						+ dstPO.getDbUserName());
				String[] srcTableConfig = baseInfoService.getDbConfig(dstPO.getDbLinkId(), dstPO.getDbUserName());
				DataSource srcTableDs = dynamicDataSourceManager.getDruidDataSource(
						dstPO.getDbLinkId() + dstPO.getDbUserName(), srcTableConfig);
				Connection srcTableCon = DataSourceUtils.getConnection(srcTableDs);
				if (DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(srcTableConfig[0].toUpperCase())) {
					srcTableCon.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
				}

				// 向从表目标表插入查询值
				for (int i = 0; i < subRelColumnsIndex.size(); i++) {
					targetPsmt_select.setObject(i + 1, targetTableRs.getString(subRelColumnsIndex.get(i)));
				}

				// 获取从表目标数据
				ResultSet subTargetRs = targetPsmt_select.executeQuery();
				ResultSetMetaData subTargetRsmd = subTargetRs.getMetaData();
				int subTargetColumns = subTargetRsmd.getColumnCount();

				// 拼接insert语句 和 delete语句
				StringBuffer srcTableSql = new StringBuffer();
				srcTableSql.append("insert into " + stsiPO.getSubSrcTableName() + "(");
				StringBuffer srcTableSql_colValue = new StringBuffer();
				StringBuffer targetTableDeleteSql = new StringBuffer();
				targetTableDeleteSql.append("delete from " + stsiPO.getSubTargetTableName() + " where 1=1");

				// 目标表需要恢复的字段的序号
				List<Integer> columnsIndex = new ArrayList<Integer>();

				// 目标表字段
				for (int j = 1; j <= subTargetColumns; j++) {
					for (TargetColMapPO po : stsiPO.getTcmPOList()) {
						if (po.getTargetColumn().toLowerCase().equals(subTargetRsmd.getColumnName(j).toLowerCase())) {// 需要恢复
							columnsIndex.add(j);
							srcTableSql.append(po.getSourceColumn() + ",");
							targetTableDeleteSql.append(" and " + po.getTargetColumn() + " = ?");
							srcTableSql_colValue.append("?,");
							break;
						}
					}
				}
				srcTableSql.deleteCharAt(srcTableSql.length() - 1);
				srcTableSql.append(")values(");
				srcTableSql_colValue.deleteCharAt(srcTableSql_colValue.length() - 1);
				srcTableSql.append(srcTableSql_colValue.toString() + ")");
				logger.info("拼接从表insert语句，任务ID=" + dmTaskId + "，sql" + srcTableSql);
				logger.info("拼接从表delete语句，任务ID=" + dmTaskId + "，sql" + targetTableDeleteSql);

				// 把要恢复的数据放到sql里
				PreparedStatement srcTablePstmt = srcTableCon.prepareStatement(srcTableSql.toString());
				PreparedStatement targetPsmt_delete = targetTableCon.prepareStatement(targetTableDeleteSql.toString());
				while (subTargetRs.next()) {
					// 向源表、目标表的sql插入数据
					for (int j = 0; j < columnsIndex.size(); j++) {
						srcTablePstmt.setObject(j + 1, targetTableRs.getString(columnsIndex.get(j)));
						targetPsmt_delete.setObject(j + 1, targetTableRs.getString(columnsIndex.get(j)));
					}
					// 从表源表插入语句加入批次
					logger.info("从表源表插入语句加入批次，任务ID=" + dmTaskId);
					srcTablePstmt.addBatch();
					// 从表目标表删除语句加入批次
					logger.info("从表目标表删除语句加入批次，任务ID=" + dmTaskId);
					targetPsmt_delete.addBatch();
				}
				logger.info("从表源表插入语句统一提交，任务ID=" + dmTaskId);
				srcTablePstmt.executeBatch();
				JdbcUtils.closeStatement(srcTablePstmt);
				logger.info("从表目标表删除语句统一提交，任务ID=" + dmTaskId);
				targetPsmt_delete.executeBatch();
				JdbcUtils.closeStatement(targetPsmt_delete);
				JdbcUtils.closeResultSet(subTargetRs);
			} catch (Exception e) {
				e.printStackTrace();
				// 从表数据恢复失败
				logger.error("数据恢复失败，任务ID=" + dmTaskId, e);
				// 更新从表任务环节表
				logger.info("更新从表任务环节表-失败，任务ID=" + dmTaskId);
				DmTaskStagePO dtsPO_update = new DmTaskStagePO();
				dtsPO_update.setDmInfoId(stsiPO.getDmInfoId());
				dtsPO_update.setState("0SF");
				dtsPO_update.setStateDate(new Date());
				dmTaskStageDao.updateByPrimaryKeySelective(dtsPO_update);
				// 更新从表任务表
				logger.info("更新从表任务表-失败，任务ID=" + dmTaskId);
				DmTaskPO dtPO_update = new DmTaskPO();
				dtPO_update.setDmTaskId(dtPO.getDmTaskId());
				dtPO_update.setState("0SF");
				dtPO_update.setStateDate(new Date());
				dmTaskDao.updateByPrimaryKeySelective(dtPO_update);
				// 更新从表恢复状态表
				logger.info("更新从表恢复状态表-失败，任务ID=" + dmTaskId);
				DmDataStatePO ddsPO_update = new DmDataStatePO();
				ddsPO_update.setDmPlanId(dtPO.getDmPlanId());
				ddsPO_update.setState(4);
				ddsPO_update.setStateDate(new Date());
				dmDataStateDao.updateByPrimaryKeySelective(ddsPO_update);
			}
			// 更新从表环节记录状态为已恢复
			logger.info("更新从表环节记录状态为已恢复，任务ID=" + dmTaskId);
			DmTaskStagePO dtsPO_update = new DmTaskStagePO();
			dtsPO_update.setDmInfoId(stsiPO.getDmInfoId());
			dtsPO_update.setState("00S");
			dtsPO_update.setStateDate(new Date());
			dmTaskStageDao.updateByPrimaryKeySelective(dtsPO_update);
		}
	}

	private String getTableName(Integer dataId, SubTableSqlInfoPO stsiPO, List<String> columnsIndex_main,
			ResultSet targetTableRs) throws SQLException {
		StringBuffer tableName = new StringBuffer();
		SubTableDefineArg stdArg = new SubTableDefineArg();
		SubTableDefineCriteria stdCriteria = stdArg.createCriteria();
		stdCriteria.andSourceDataIdEqualTo(dataId);
		stdCriteria.andStateEqualTo(DMSConstant.Status.SUCCESS.toString());
		stdArg.setOrderByClause("priority");
		List<SubTableDefinePO> stdPOList_src = subTableDefineDao.selectByArg(stdArg);
		for (SubTableDefinePO stdPO : stdPOList_src) {
			if (stdPO.getPriority() == 1) {// 基表名
				tableName.append(stdPO.getSubDefineValue());
				continue;
			}
			if (stdPO.getSubDefineType() == DMSConstant.SUB_DEFINE_TYPE_FIXED) {// 固定部分
				tableName.append("_" + stdPO.getSubDefineValue());
			} else if (stdPO.getSubDefineType() == DMSConstant.SUB_DEFINE_TYPE_CHANGE) {// 可变
				// 获取字段对应的值
				Object subRefColValue = null;
				// 先找出该主表源表字段对应的目标表的字段
				String targetColumn = "";
				for (TargetColMapPO tcmPO : stsiPO.getTcmPOList()) {
					if (tcmPO.getSourceColumn().equals(stdPO.getSubRefCol())) {
						targetColumn = tcmPO.getTargetColumn();
						break;
					}
				}
				for (int i = 0; i < columnsIndex_main.size(); i++) {
					if (columnsIndex_main.get(i).split(",")[1].equals(targetColumn)) {
						subRefColValue = targetTableRs.getString(Integer
								.parseInt(columnsIndex_main.get(i).split(",")[0]));
						break;
					}
				}

				switch (Integer.parseInt(stdPO.getSubDefineValue())) {
				case 2:
					stsiPO.setSubSrcTableName("_" + exchangeFunc.yearMonthExchange((Date) subRefColValue));
					tableName.append("_" + exchangeFunc.yearMonthExchange((Date) subRefColValue));
					break;
				case 3:
					tableName.append("_" + exchangeFunc.dateExchange((Date) subRefColValue));
					break;
				case 4:
					tableName.append("_" + exchangeFunc.billingCycleExchange((Date) subRefColValue));
					break;
				default:// 原值
					tableName.append("_" + String.valueOf(subRefColValue));
					break;
				}
			}
		}
		return tableName.toString();
	}
}
